package com.zjc.mymq.broker.event.spi.listener;

import org.apache.commons.collections4.CollectionUtils;
import com.zjc.mymq.broker.cache.CommonCache;
import com.zjc.mymq.broker.core.CommitLogMMapFileModel;
import com.zjc.mymq.broker.event.model.TimeWheelEvent;
import com.zjc.mymq.broker.timewheel.SlotStoreTypeEnum;
import com.zjc.mymq.broker.timewheel.TimeWheelSlotModel;
import com.zjc.mymq.common.dto.ConsumeMsgCommitLogDTO;
import com.zjc.mymq.common.dto.MessageDTO;
import com.zjc.mymq.common.dto.MessageRetryDTO;
import com.zjc.mymq.common.event.Listener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

/**
 * @author zjc
 * @date Created at 2024/8/4
 *  时间轮监听器
 */
public class TimeWheelListener implements Listener<TimeWheelEvent> {

    private final Logger logger = LoggerFactory.getLogger(TimeWheelListener.class);

    @Override
    public void onReceive(TimeWheelEvent event) throws Exception {
        List<TimeWheelSlotModel> timeWheelSlotModelList = event.getTimeWheelSlotModelList();
        if(CollectionUtils.isEmpty(timeWheelSlotModelList)) {
            logger.error("timeWheelSlotModelList is empty");
            return;
        }
        for (TimeWheelSlotModel timeWheelSlotModel : timeWheelSlotModelList) {
            if (SlotStoreTypeEnum.MESSAGE_RETRY_DTO.getClazz().equals(timeWheelSlotModel.getStoreType())) {
                MessageRetryDTO messageRetryDTO = (MessageRetryDTO) timeWheelSlotModel.getData();
                this.messageRetryHandler(messageRetryDTO);
            }
        }
    }

    /**
     * 消息重试处理器
     *
     * @param messageRetryDTO
     */
    private void messageRetryHandler(MessageRetryDTO messageRetryDTO) {
        //将消息数据扔到需要重试的topic里
        CommitLogMMapFileModel commitLogMMapFileModel = CommonCache.getCommitLogMMapFileModelManager().get(messageRetryDTO.getTopic());
        ConsumeMsgCommitLogDTO consumeMsgCommitLogDTO = commitLogMMapFileModel.readContent(messageRetryDTO.getSourceCommitLogOffset(),messageRetryDTO.getSourceCommitLogSize());
        byte[] commitLogBody = consumeMsgCommitLogDTO.getBody();
        System.out.println("扔到重试topic的数据：" + new String(commitLogBody));
        MessageDTO messageDTO = new MessageDTO();
        messageDTO.setBody(commitLogBody);
        messageDTO.setTopic("retry%" + messageRetryDTO.getConsumeGroup());
        messageDTO.setQueueId(ThreadLocalRandom.current().nextInt(3));
        messageDTO.setCurrentRetryTimes(messageRetryDTO.getCurrentRetryTimes() + 1);
        logger.info("重试次数增加：{}",messageDTO.getCurrentRetryTimes());
        try {
            CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
